package com.rxjava;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

//http://reactivex.io/documentation/operators.html
public class SimpleDemo {

	/**先创建个数据发射源,很好理解，就是发射数据用的：
	 * Observable：发射源，英文释义“可观察的”，在观察者模式中称为“被观察者”或“可观察对象”；
	 * Subscriber：“订阅者”，也是接收源，那它跟Observer有什么区别呢？Subscriber实现了Observer接口，比Observer多了一个最重要的方法unsubscribe( )，用来取消订阅，当你不再想接收数据了，可以调用unsubscribe( )方法停止接收，Observer 在 subscribe() 过程中,最终也会被转换成 Subscriber 对象，一般情况下，建议使用Subscriber作为接收源；
	 * @return
	 */
	public Observable<Integer> send(int count) {
		Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			/**
			 * ObservableEmitter： Emitter是发射器的意思，那就很好猜了，这个就是用来发出事件的，它可以发出三种类型的事件，通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。
			但是，请注意，并不意味着你可以随意乱七八糟发射事件，需要满足一定的规则：
			上游可以发送无限个onNext, 下游也可以接收无限个onNext.
			当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
			当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
			上游可以不发送onComplete或onError.
			最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
			注: 关于onComplete和onError唯一并且互斥这一点, 是需要自行在代码中进行控制, 如果你的代码逻辑中违背了这个规则, 并不一定会导致程序崩溃. 比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了, 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃.
			 * */
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				for (int i = 0; i < count; i++) {
					emitter.onNext(i);
				}
				emitter.onComplete();
			}
		});
		return observable;
	}

	/**再创建个数据接收源，同理，接收数据用的：
	 * Observer：接收源，英文释义“观察者”，没错！就是观察者模式中的“观察者”，可接收Observable、Subject发射的数据；
	 * @return
	 */
	public Observer<Integer> revieve() {
		// 创建一个下游 Observer
		Observer<Integer> observer = new Observer<Integer>() {
			@Override
			public void onSubscribe(Disposable d) {
				System.out.println("subscribe");
			}

			@Override
			public void onNext(Integer value) {
				System.out.println(value);
			}

			@Override
			public void onError(Throwable e) {
				System.out.println("error");
			}

			@Override
			public void onComplete() {
				System.out.println("complete");
			}
		};
		return observer;
	}

	/**注意: 只有当上游和下游建立连接之后, 上游才会开始发送事件. 也就是调用了subscribe() 方法之后才开始发送事件.
	 * @param args
	 */
	public static void main(String[] args) {
		SimpleDemo demo = new SimpleDemo();
		//将发射源和接收源关联起来：
		demo.send(3).subscribe(demo.revieve());

		//把这段代码连起来写就成了RxJava引以为傲的链式操作：
		Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            	System.out.println("emit 1");
                emitter.onNext(1);
                System.out.println("emit 2");
                emitter.onNext(2);
                System.out.println("emit 3");
                emitter.onNext(3);
                System.out.println("emit complete");
                emitter.onComplete();
                System.out.println("emit 4");
                emitter.onNext(4);
                System.out.println("emit 5");
                emitter.onNext(5);
            }
        }).subscribe(new Observer<Integer>() {
        	private Disposable mDisposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
            	System.out.println("subscribe");
            	mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
            	System.out.println(value);
            	//调用dispose()并不会导致上游不再继续发送事件, 用于解除订阅。上游会继续发送剩余的事件。
            	//从运行结果我们看到, 在收到onNext 2这个事件后, 切断了水管, 但是上游仍然发送了3, complete, 4, 5这几个事件, 而且上游并没有因为发送了onComplete而停止. 同时可以看到下游的onSubscribe()方法是最先调用的。
            	i++;
                if (i == 2) {
                	System.out.println("dispose");
                    mDisposable.dispose();
                    System.out.println("isDisposed : " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
            	System.out.println("error");
            }

            @Override
            public void onComplete() {
            	System.out.println("complete");
            }
        });
	}

	/**
	 * Action0：RxJava中的一个接口，它只有一个无参call（）方法，且无返回值，同样还有Action1，Action2...Action9等，Action1封装了含有 1 个参的call（）方法，即call（T t），Action2封装了含有 2 个参数的call方法，即call（T1 t1，T2 t2），以此类推；
	 * Func0：与Action0非常相似，也有call（）方法，但是它是有返回值的，同样也有Func0、Func1...Func9;
	 * */

}
